#pragma once

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../common/Log.hpp"
#include "../common/UUIDHelper.hpp"
#include "../common/message.pb.h"
#include "../common/protocol.pb.h"
#include "Consumer.hpp"
#include <string>
#include <unordered_map>
#include <memory>
#include <vector>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <cstdio>

namespace micromq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

    using ConsumerResponsePtr = std::shared_ptr<ConsumerResponse>;
    using CommonResponsePtr = std::shared_ptr<CommonResponse>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
    public:
        Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
            : _channelId(UUIDHelper::getUUID()), _conn(conn), _codec(codec)
        {}

        bool openChannel()
        {
            OpenChannelRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            _codec->send(_conn, req);
            CommonResponsePtr resp = waitResponse(reqId);
            logInfo("open channel %s...", resp->is_ok() ? "success" : "faild");
            return resp->is_ok();
        }

        void closeChannel()
        {
            CloseChannelRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            _codec->send(_conn, req);
            waitResponse(reqId);
            logInfo("close channel success...");
        }

        bool declareExchange(const std::string& name, ExchangeType type, bool durable, bool autoDelete, 
            google::protobuf::Map<std::string, std::string>& args)
        {
            DeclareExchangeRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_exchange_name(name);
            req.set_echange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(autoDelete);
            req.mutable_args()->swap(args);
            _codec->send(_conn, req); //发送请求
            CommonResponsePtr resp = waitResponse(reqId); //等待服务端响应
            logInfo("declare exchange: %s %s...", name.c_str(), resp->is_ok() ? "success" : "faild");
            return resp->is_ok();
        }

        void removeExchange(const std::string& name)
        {
            RemoveExchangeRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_exchange_name(name);
            _codec->send(_conn, req);
            waitResponse(reqId);
            logInfo("remove exchange: %s success...", name.c_str());
        }

        bool declareMagQueue(const std::string& name, bool durable, bool exclusive, bool autoDelete, 
            google::protobuf::Map<std::string, std::string>& args)
        {
            DeclareMsgQueueRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_msg_queue_name(name);
            req.set_durable(durable);
            req.set_exclusive(exclusive);
            req.set_auto_delete(autoDelete);
            req.mutable_args()->swap(args);
            _codec->send(_conn, req);
            CommonResponsePtr resp = waitResponse(reqId);
            logInfo("declare message queue: %s %s...", name.c_str(), resp->is_ok() ? "success" : "faild");
            return resp->is_ok();
        }

        void removeMsgQueue(const std::string& name)
        {
            RemoveMsgQueueRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_msg_queue_name(name);
            _codec->send(_conn, req);
            waitResponse(reqId);
            logInfo("remove message queue: %s success...", name.c_str());
        }

        bool bind(const std::string& exchangeName, const std::string& msgQueueName, const std::string& bindingKey)
        {
            BindRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_exchange_name(exchangeName);
            req.set_msg_queue_name(msgQueueName);
            req.set_binding_key(bindingKey);
            _codec->send(_conn, req);
            CommonResponsePtr resp = waitResponse(reqId);
            logInfo("bind %s with %s %s...", exchangeName.c_str(), msgQueueName.c_str(), resp->is_ok() ? "success" : "faild");
            return resp->is_ok();
        }

        void unbind(const std::string& exchangeName, const std::string& msgQueueName)
        {
            UnbindRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_exchange_name(exchangeName);
            req.set_msg_queue_name(msgQueueName);
            _codec->send(_conn, req);
            waitResponse(reqId);
            logInfo("unbind %s with %s success...", exchangeName.c_str(), msgQueueName.c_str());
        }

        void publishMessage(const std::string& exchangeName, BasicProperties* bp, const std::string& content)
        {
            PublishMessageRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_exchange_name(exchangeName);
            if(bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_mode(bp->mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            req.set_content(content);
            _codec->send(_conn, req);
            waitResponse(reqId);
            logInfo("publish message: %s to exchang: %s success...", content.c_str(), exchangeName.c_str());
        }

        void ackMessage(const std::string& msgId)
        {
            if(_consumer.get() == nullptr)
            {
                logInfo("consumer can not found...");
                return;
            }
            AckMessageRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_msg_queue_name(_consumer->_msgQueueName);
            req.set_message_id(msgId);
            _codec->send(_conn, req);
            waitResponse(reqId);
            logInfo("ack message: %s success...", msgId.c_str());
        }

        bool followMsgQueue(const std::string& tag, const std::string& msgQueueName, bool autoAck, ConsumerCallback cb)
        {
            if(_consumer.get() != nullptr)
            {
                logInfo("follow message queue faild, consumer has followed another message queue...");
                return false;
            }
            FollowMsgQueueRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_consumer_tag(tag);
            req.set_msg_queue_name(msgQueueName);
            req.set_auto_ack(autoAck);
            _codec->send(_conn, req);
            CommonResponsePtr resp = waitResponse(reqId);
            if(resp->is_ok() == false)
            {
                logInfo("follow message queue faild...");
                return false;
            }
            _consumer = std::make_shared<Consumer>(tag, msgQueueName, autoAck, cb);
            logInfo("follow message queue: %s %s...", msgQueueName.c_str(), resp->is_ok() ? "success" : "faild");
            return true;
        }

        void unfollowMsgQueue()
        {
            if(_consumer.get() == nullptr) return;
            UnfollowMsgQueueRequest req;
            std::string reqId = UUIDHelper::getUUID();
            req.set_req_id(reqId);
            req.set_channel_id(_channelId);
            req.set_consumer_tag(_consumer->_tag);
            req.set_msg_queue_name(_consumer->_msgQueueName);
            _codec->send(_conn, req);
            waitResponse(reqId);
            _consumer.reset(); //智能指针引用计数--
            logInfo("unfollow message queue success...");
        }

        void putResponse(const CommonResponsePtr& resp)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _respMap[resp->resp_id()] = resp;
            _cond.notify_all();
        }

        void consumeMessage(const ConsumerResponsePtr& resp)
        {
            if(_consumer.get() == nullptr)
            {
                logInfo("consumer can not found...");
                return;
            }
            if(_consumer->_tag != resp->consumer_tag())
            {
                logInfo("consumer tag match faild: %s != %s", _consumer->_tag.c_str(), resp->consumer_tag().c_str());
                return;
            }
            _consumer->_callback(resp->consumer_tag(), resp->mutable_properties(), resp->content());
        }

        std::string getChannelId()
        {
            return _channelId;
        }

        ~Channel()
        {
            unfollowMsgQueue();
        }
    private:
        CommonResponsePtr waitResponse(const std::string& reqId)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _cond.wait(lock, [this, &reqId]() {
                return _respMap.find(reqId) != _respMap.end();
            });
            CommonResponsePtr resp = _respMap[reqId];
            _respMap.erase(reqId);
            return resp;
        }
    private:
        std::string _channelId;
        Consumer::ptr _consumer;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        std::mutex _mtx;
        std::condition_variable _cond;
        std::unordered_map<std::string, CommonResponsePtr> _respMap; // <响应id，响应指针>
    };
}